package to_hbase_xiemingyang

import org.apache.hadoop.hbase.client.{ConnectionFactory, Put, Table}
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}

import scala.io.Source

object make_map01_map02_to_hb {
  val hbase_zq = "10.120.65.181,10.120.67.164,10.120.64.243,10.120.64.144"
  val hbase_zqPort = "2181"
  val hBase_zParent = "/hbase"
  val configuration = HBaseConfiguration.create()
  this.configuration.set("hbase.zookeeper.quorum", this.hbase_zq)
  this.configuration.set("hbase.zookeeper.property.clientPort", this.hbase_zqPort)
  this.configuration.set("zookeeper.znode.parent", this.hBase_zParent) //some condition you should add;
  val connection = ConnectionFactory.createConnection(this.configuration)
  val admin = connection.getAdmin()

  def curTable(tablename: String): Table = {
    val tName = TableName.valueOf(tablename)
    this.connection.getTable(tName)
  }

  /**
   * 向HBase中插入一条数据
   *
   * @param tableNameStr
   * @param datas 一个Array数组,其中datas[0]是RowKey,datas[2]是列族,datas[3]是列名,datas[4]是列值
   */
  def putData(tableNameStr: String, datas: Array[String]) = {
    val table: Table = curTable(tableNameStr)
    val cell: Array[String] = datas
    val put: Put = new Put(Bytes.toBytes(cell(0)))
    put.addColumn(Bytes.toBytes(cell(1)), Bytes.toBytes(cell(2)), Bytes.toBytes(cell(3)))
    table.put(put)
    table.close()
  }

  /**
   * 将一个文件的所有行插入到HBase中
   *
   * @param tableNameStr 表名
   * @param datas        一个文件的所有行,是一个Array,每行是一个字符串元素
   */
  def putDatas(tableNameStr: String, datas: Array[String]) = {
    val table: Table = curTable(tableNameStr)
    try {
      for (data <- datas) {
        val cell: Array[String] = data.split("\t")
        if (cell.length < 4) {
          println(1)
        }

        //单个sourcecode
        /*val rowkey = "UNIDO_" + cell(1)
        val put: Put = new Put(Bytes.toBytes(rowkey))
        put.addColumn(Bytes.toBytes(cell(0)), Bytes.toBytes(cell(2)), Bytes.toBytes(cell(3)))
        table.put(put)*/
        //针对多SourceCode的情况
        //表:"BaseMapInfos"
        //rowkey:webcode(写死)_sourcecode
        //列族:map01或map02或dealTypes
        //列族:map01 列名:kbase列名,列值:value表列名
        //列族:map02 列名value表的列名,列值:concept表的conceptId
        //列族:dealTypes 列名:kbase表的列名,列值对应type类型
        val sourcecodes: Array[String] = cell(1).split(",")
        sourcecodes.foreach(sourcecode => {
          val rowkey = "IBRD_" + sourcecode
          val put: Put = new Put(Bytes.toBytes(rowkey))
          ////addColumn(byte [] family, byte [] qualifier, byte [] value)
          put.addColumn(Bytes.toBytes(cell(0)), Bytes.toBytes(cell(2)), Bytes.toBytes(cell(3)))
          table.put(put)
        })
      }
    } catch {
      case e: Exception => e.printStackTrace()
    } finally {
      table.close()
    }
  }

  def main(args: Array[String]): Unit = {
    val path = make_map01_map02_to_hb.getClass.getClassLoader().getResource("Map01_02_dealTypesList/IBRD_map01_dealTypes.txt").getPath
    val source = Source.fromFile(path, "utf-8")
    val lines = source.getLines().toArray
    source.close()
    val tablename = "BaseMapInfos"
    putDatas(tablename, lines)

  }

}

